package cn.nawang.ebeim.server.service;

import cn.nawang.ebeim.server.constants.Config;
import cn.nawang.ebeim.server.constants.Constant;
import cn.nawang.ebeim.server.entity.StorageServer;
import cn.nawang.ebeim.server.util.ApiVisitUtil;
import cn.nawang.ebeim.server.util.SystemUtil;
import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 封装zookeeper客户端
 * 提供zookeeper注册，节点查询操作
 * (主要是为了解决之前存在的递归创建连接的弊端)
 * Created by GanJc on 2015-11-20 08:56
 */
public class ZookeeperService implements Watcher{

    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperService.class);

    private static ZookeeperService zookeeperService = new ZookeeperService();

    private ZooKeeper zooKeeper;

    private CountDownLatch countDownLatch ;

    private ZookeeperService() {
        init();
    }

    public static ZookeeperService instance() {
        return zookeeperService;
    }

    /**
     *  每次调用初始化init(),则释放旧连接内存资源，
     *  创建新的zookeeper连接
     */
    private void init() {
        LOG.info("zookeeper init!");
        try {
            if (zooKeeper != null) {
                zooKeeper.close();
                zooKeeper = null ;
            }
            countDownLatch = new CountDownLatch(1);
            String zkHosts = ApiVisitUtil.getZookeeperHosts();
//            String zkHosts = "121.41.106.249:2181"; //test
            int sessionTime = Config.SESSION_TIME;
            zooKeeper = new ZooKeeper(zkHosts, sessionTime, this);
            countDownLatch.await();
        } catch (Exception e) {
            LOG.info("zookeeper服务器不可用：5秒后重试!" + e);
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            init();
        }finally {
            countDownLatch = null ;
        }
    }

    @Override
    public void process(WatchedEvent event) {
        switch (event.getState()) {
            case SyncConnected:
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                break;
            case Expired:
                LOG.info("zookeeper session expire,reconnect!" );
                init();
                register();
                break;
        }
    }

    /**
     * 获取可用存储云服务器列表
     * @param storageCloudId
     * @return
     */
    public List<StorageServer> getStorageServers(String storageCloudId) {
        List<StorageServer> servers = null;
        try {
            List<String> list = getChildNodeList(Constant.ZK_GROUP_ROOT);
            // 获取到Zookeeper的根目录
            if (list != null && list.size() > 0) {
                servers = new ArrayList<StorageServer>();
                for (int i = 0; i < list.size(); i++) {
                    String path = Constant.ZK_GROUP_ROOT + "/" + list.get(i);
                    String serverInfo = getData(path);
                    String[] arrInfo = serverInfo.split(":");
                    String ip = arrInfo[0];
                    int port = Integer.valueOf(arrInfo[1]);
                    servers.add(new StorageServer(ip, port));
                }
            }
            return servers;
        } catch (Exception e) {
            e.printStackTrace();
            //zookeeper未连接，无法获取serverList,下次再获取即可，
            return null;
        }
    }

    public void register() {
        try {
            if (!exists(Constant.ZK_GROUP_ROOT)) {
                create(Constant.ZK_GROUP_ROOT);
            }
            String nodePath = Constant.ZK_GROUP_ROOT + "/" + SystemUtil.getLocalHostName();
            if (exists(nodePath)) {
                LOG.info("Already exist, do nothing. {}", nodePath);
                return ;
            }
            createNodeWithContent(nodePath);
            LOG.info("register zookeeper success - {}", nodePath);
        } catch (KeeperException e) {
            LOG.info("zookeeper session expire," + e);
            init();
            register();
        } catch (Exception e){
            throw new RuntimeException(e);
        }
    }

    private List<String> getChildNodeList(String path)  {
        List<String> list = null;
        try {
            list = zooKeeper.getChildren(path, false);
            if (list.isEmpty()) {
                LOG.info(path + " 中没有节点");
            } else {
                LOG.info(path + " 中存在节点");
            }
        } catch (KeeperException e) {
            LOG.info("zookeeper session expire," + e);
            init();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return list ;
    }

    private void create(String path, byte[] data) throws KeeperException, InterruptedException {
        zooKeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    }

    private void create(String path) throws KeeperException, InterruptedException {
        zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    private boolean exists(String nodePath) throws KeeperException, InterruptedException {
        if (zooKeeper.exists(nodePath, true) != null) {
            return true;
        }
        return false;
    }

    private void createNodeWithContent(String path) throws KeeperException, InterruptedException {
        //获取节点数据 改为从配置文件里取 ，在那台服务器启动server 就在哪台服务器的配置文件 配置该server的 外网地址
        String nodeContent = Config.UPLOAD_HOST + ":" + Config.UPLOAD_PORT;
        if (Config.SERVER_HOST != null) {
            nodeContent = Config.SERVER_HOST + ":" + Config.UPLOAD_PORT;
        }
        //创建节点和节点数据
        create(path, nodeContent.getBytes());
        //监听此节点
        exists(path);
    }

    private String getData(String path) throws KeeperException, InterruptedException {
        return new String(zooKeeper.getData(path, false, null));
    }

}
